package com.arelikebrothers.demo.kafka.stream;

import com.arelikebrothers.demo.kafka.stream.bean.InputMessage;
import com.arelikebrothers.demo.kafka.stream.bean.OutputMessage;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.util.Properties;
import java.util.concurrent.Executors;

/**
 * Created by lennon on 16/07/2017.
 */
public class KafkaStreamsDemo {

    private Properties config = new Properties();
    private KStream<String, InputMessage> source;
    private KStreamBuilder builder;


    public KafkaStreamsDemo() {
        initConfig();
        buildKafkaStream();
    }

    public void start() {
        Executors.newSingleThreadExecutor(r -> {
            Thread thread = new Thread(r, "KAFKA-STREAMS-THREAD");
            thread.setDaemon(true);
            return thread;
        }).submit(() -> {
            KafkaStreams streams = new KafkaStreams(builder, config);
            streams.start();
        });
    }

    private void buildKafkaStream() {
        builder = new KStreamBuilder();
//        builder = new TopologyBuilder();
//        builder.addSource("SOURCE", "INPUT")
//                .addProcessor("CALC", DemoProcessor::new, "SOURCE")
//                .addStateStore(Stores.create("RESULT")
//                                .withStringKeys()
//                                .withStringValues()
//                                .inMemory()
//                                .build(),
//                        "CALC")
//                .addSink("SINK", "OUTPUT_RESULT", "CALC");


        source = builder.stream(DemoSerdes.String(), DemoSerdes.Json(InputMessage.class), "INPUT");

        KStream<String, OutputMessage> messageKStream = source.map((key, value) -> {
            if (value == null) {
                return new KeyValue<>(null, null);
            }

            OutputMessage outputMessage = new OutputMessage(value.getKey1() + value.getKey2(), value.getContent());
            return new KeyValue<>(outputMessage.getId(), outputMessage);

        }).filter((k, v) -> k != null && v != null);
        messageKStream.print();
//        KTable<String, OutputMessage> xxxx = messageKStream.groupByKey().aggregate(new Initializer<OutputMessage>() {
//            @Override
//            public OutputMessage apply() {
//                return new OutputMessage();
//            }
//        }, new Aggregator<String, OutputMessage, OutputMessage>() {
//            @Override
//            public OutputMessage apply(String s, OutputMessage v, OutputMessage output) {
//                output.setId(v.getId());
//                output.setValue(v.getValue());
//                return output;
//            }
//        }, DemoSerdes.Json(OutputMessage.class), "xxxx");
//        xxxx.toStream().print();

//        KTable<Windowed<String>, OutputMessage> timewindow = messageKStream.aggregateByKey(() -> new OutputMessage(), new Aggregator<String, OutputMessage, OutputMessage>() {
//                    @Override
//                    public OutputMessage apply(String s, OutputMessage v, OutputMessage output) {
//                        output.setId(v.getId());
//                        output.setValue(v.getValue());
//                        return output;
//                    }
//                },
//                TimeWindows.of("timewindow", 1000),
//                DemoSerdes.String(),
//                DemoSerdes.Json(OutputMessage.class));
//
//        timewindow.to(DemoSerdes.Window(DemoSerdes.String()), DemoSerdes.Json(OutputMessage.class),
//                (k, v, c) -> 0, "OUTPUT_TOPIC");
//        timewindow.print(DemoSerdes.Window(DemoSerdes.String()), DemoSerdes.Json(OutputMessage.class));

    }

    private void initConfig() {
        config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app-demo1");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(StreamsConfig.CLIENT_ID_CONFIG, "STREAM-CONSUMER1");
        config.put(StreamsConfig.STATE_DIR_CONFIG, "/opt/kafka/tmp");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    }

    public static void main(String[] args) throws InterruptedException {
        KafkaStreamsDemo demo = new KafkaStreamsDemo();
        demo.start();

        Thread.sleep(5000000);
    }
}
